package com.codetile.batch.execution;

import com.codetile.batch.db.User;
import org.springframework.batch.item.*;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.mapping.DefaultLineMapper;
import org.springframework.batch.item.file.mapping.FieldSetMapper;
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
import org.springframework.batch.item.file.transform.FieldSet;
import org.springframework.core.io.ClassPathResource;
import org.springframework.stereotype.Component;
import org.springframework.validation.BindException;

@Component("streamReader")
public class FileItemReader implements ItemStreamReader<User> {

    private FlatFileItemReader<User> userFlatFileItemReader=new FlatFileItemReader<>();
    private Long curLine=0L;
    private boolean restart=false;
    private ExecutionContext executionContext;

    public FileItemReader(){
        userFlatFileItemReader.setResource(new ClassPathResource("user.txt"));
        userFlatFileItemReader.setLinesToSkip(1);
        DelimitedLineTokenizer tokenizer=new DelimitedLineTokenizer();
        tokenizer.setNames(new String[]{"id","name","age"});
        DefaultLineMapper<User> mapper=new DefaultLineMapper<>();
        mapper.setLineTokenizer(tokenizer);
        mapper.setFieldSetMapper(new FieldSetMapper<User>() {
            @Override
            public User mapFieldSet(FieldSet fieldSet) throws BindException {
                User user=new User();
                user.setId(fieldSet.readInt("id"));
                user.setName(fieldSet.readString("name"));
                user.setAge(fieldSet.readInt("age"));
                return user;
            }
        });

        mapper.afterPropertiesSet();
        userFlatFileItemReader.setLineMapper(mapper);
    }

    @Override
    public User read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
        User user=null;
        this.curLine++;

        if(restart){
            userFlatFileItemReader.setLinesToSkip(this.curLine.intValue()-1);
            restart=false;
            System.out.println("start reading from line:"+this.curLine);
        }
        userFlatFileItemReader.open(this.executionContext);
        user=userFlatFileItemReader.read();

        if(user!=null&&user.getName().equals("麻子")){
            throw new RuntimeException("something wrong ,user id:"+user.getId());
        }

        return user;
    }

    @Override
    public void open(ExecutionContext executionContext) throws ItemStreamException {
        this.executionContext=executionContext;
        if(executionContext.containsKey("curLine")){
            this.curLine=executionContext.getLong("curLine");
            this.restart=true;
        }else {
            this.curLine=0L;
            executionContext.put("curLine",this.curLine);
            System.out.println("start reading form line:"+this.curLine+1);
        }
    }

    @Override
    public void update(ExecutionContext executionContext) throws ItemStreamException {
        executionContext.put("curLine",this.curLine);
        System.out.println("currentLine:"+this.curLine);
    }

    @Override
    public void close() throws ItemStreamException {

    }
}
